/*
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.facebook.presto.plugin.neo4j.adaptive;

import com.facebook.airlift.log.Logger;
import com.facebook.presto.plugin.neo4j.Neo4jConfig;
import com.facebook.presto.spi.PrestoException;
import com.facebook.presto.spi.adaptive.metadata.AdaptiveConnectorMetadaManager;
import com.google.common.annotations.VisibleForTesting;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;

import java.util.List;

import static com.facebook.presto.plugin.neo4j.Neo4jErrorCode.METADATA_ERROR;

public class NMetadataManager
        implements AdaptiveConnectorMetadaManager
{
    private static final Logger log = Logger.get(NMetadataManager.class);

    private String ZK_HOST;
    private String NEO4J_METADATA_PATH;
    private String adptiveEnable;

    private CuratorFramework client = null;
    private NMetadata nMetadata;

    public NMetadataManager(Neo4jConfig neo4jConfig)
    {
        this.adptiveEnable = neo4jConfig.getAdaptiveEnable();
        this.ZK_HOST = neo4jConfig.getMetadataZKHost();
        this.NEO4J_METADATA_PATH = neo4jConfig.getMetadataZKNode();

        if (isEnable()) {
            client = CuratorFrameworkFactory
                    .builder()
                    .connectString(ZK_HOST)
                    .retryPolicy(new ExponentialBackoffRetry(1000, 3))
                    .build();
            client.start();

            this.nMetadata = readNeo4jMetadataFromZK();
        }
    }

    public boolean isEnable()
    {
        return adptiveEnable.trim().equals("true");
    }

    /**
     * Add a table to Neo4j and write metadata to zk.
     */
    public void addATable(NTable nTable)
    {
        nMetadata.getnTableList().add(nTable);
        writeNeo4jMetadataToZK(nMetadata);
    }

    /**
     * remove a table from Neo4j and write metadata to zk.
     */
    public void removeATable(String tableName)
    {

        nMetadata.getnTableList().remove(new NTable(tableName));
        writeNeo4jMetadataToZK(nMetadata);
    }

    /**
     * Read data and deserialize Neo4jMetadata from zk.
     * If zkNode is null, it will return a neo4jMetadata including empty NTable list.
     */
    private NMetadata readNeo4jMetadataFromZK()
    {
        try {
            return NMetadataUtil.deserialize(client.getData().forPath(NEO4J_METADATA_PATH));
        }
        catch (Exception e) {
            log.error("Neo4jMetadata read from ZK failed, " + e.getMessage());
            return new NMetadata();
        }
    }

    /**
     * Serialize Neo4jMetadata and write data to zk.
     * If zkNode is null, it will create a node and write it.
     */
    private void writeNeo4jMetadataToZK(NMetadata nMetadata)
    {
        try {

            if (client.checkExists().forPath(NEO4J_METADATA_PATH) == null) {
                client.create().creatingParentsIfNeeded().forPath(NEO4J_METADATA_PATH, NMetadataUtil.serialize(nMetadata));
            }
            else {
                client.setData().forPath(NEO4J_METADATA_PATH, NMetadataUtil.serialize(nMetadata));
            }
        }
        catch (Exception e) {
            log.error("Neo4jMetadata write to ZK failed, " + e.getMessage());
        }
    }

    /**
     * Check whether has the table.
     */
    public boolean hasTable(String tableName)
    {
        return
                nMetadata.getnTableList().stream()
                        .anyMatch(nTable -> nTable.getTableName().equals(tableName));
    }

    public NColumn getColumn(String tableName, String columnName)
    {
        return
                nMetadata.getnTableList().stream()
                        .filter(table -> table.getTableName().equals(tableName))
                        .findFirst()
                        .get()
                        .getnColumnList()
                        .stream()
                        .filter(column -> column.getColumnName().equals(columnName))
                        .findFirst()
                        .orElse(null);
    }

    public List<NColumn> getColumnList(String tableName)
    {
        return
                nMetadata.getnTableList().stream()
                        .filter(table -> table.getTableName().equals(tableName))
                        .findFirst()
                        .get()
                        .getnColumnList();
    }

    public void dropTable(String tableName)
            throws PrestoException
    {
        try {
            NTable nTable = nMetadata.getnTableList().stream()
                    .filter(table -> table.getTableName().equals(tableName))
                    .findFirst()
                    .orElse(null);
            nMetadata.getnTableList().remove(nTable);
        }
        catch (Exception e) {
            throw new PrestoException(METADATA_ERROR, "neo4j drop a table from metadataManager failed", e);
        }
    }

    @VisibleForTesting
    public NMetadataManager(String zkHost, String zkNode)
    {
        this.ZK_HOST = zkHost;
        this.NEO4J_METADATA_PATH = zkNode;
        this.adptiveEnable = "true";

        if (isEnable()) {
            client = CuratorFrameworkFactory
                    .builder()
                    .connectString(ZK_HOST)
                    .retryPolicy(new ExponentialBackoffRetry(1000, 3))
                    .build();
            client.start();

            this.nMetadata = readNeo4jMetadataFromZK();
        }
    }
}
